package com.shujia.jinjie;

import com.shujia.utils.MyHBaseTools;
import javafx.scene.control.Tab;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import redis.clients.jedis.Jedis;

import java.util.ArrayList;
import java.util.List;

/*
    将原表中的性别取出来
    男:1001,1002,....
    女:1004,1011,...

    redis的value数据类型：
    String,list,set,zset,hash

 */
public class HBaseWithRedisIndex {
    private static Connection conn;
    private static Admin admin;
    private static Jedis jedis;

    public static void main(String[] args) throws Exception{
        init();


//        TableName students2 = TableName.valueOf("students2");
//
//        //从原表中查询每个性别的学号
//        List<String> boyIdList = getBoyIds(students2);
//        List<String> girlIdList = getGirlIds(students2);
//
//        //构建索引表，在redis中构建
//        createIndexWithRedis(boyIdList,girlIdList);

        //查询数据
        getBoyData("男");
        System.out.println("===============================================");
        getGirlData("女");



    }

    public static void getGirlData(String gender) throws Exception{
        //先查询redis
        List<String> ids = jedis.lrange("gender:" + gender, 0, -1);
        Table students2 = conn.getTable(TableName.valueOf("students2"));

        //经过分析后发现，虽然我们想让hbase直接扫描一批数据，但是，得到的范围在hbase表中既有男生也有女生，我们就会想着使用过滤器，
        //但是，可笑的是，一旦用了过滤器，那我们还使用redis索引干嘛呢
        //这里使用二级索引的目的就是为了挨个通过rowkey获取数据
        for (String id : ids) {
            Get get = new Get(Bytes.toBytes(id));
            Result result = students2.get(get);
            MyHBaseTools.printResult(result);
        }
    }

    public static void getBoyData(String gender) throws Exception{
        //先查询redis
        List<String> ids = jedis.lrange("gender:" + gender, 0, -1);
        Table students2 = conn.getTable(TableName.valueOf("students2"));

        //经过分析后发现，虽然我们想让hbase直接扫描一批数据，但是，得到的范围在hbase表中既有男生也有女生，我们就会想着使用过滤器，
        //但是，可笑的是，一旦用了过滤器，那我们还使用redis索引干嘛呢
        //这里使用二级索引的目的就是为了挨个通过rowkey获取数据
        for (String id : ids) {
            Get get = new Get(Bytes.toBytes(id));
            Result result = students2.get(get);
            MyHBaseTools.printResult(result);
        }
    }

    public static void createIndexWithRedis(List<String> boyIdList,List<String> girlIdList){
        // 为了后面可以进行使用Scan范围查询，用list的原因是有序
        for (String boyId : boyIdList) {
            jedis.rpush("gender:男",boyId);
        }

        for (String girlId : girlIdList) {
            jedis.rpush("gender:女",girlId);
        }
    }

    public static List<String> getBoyIds(TableName tableName) throws Exception{
        Table students2 = conn.getTable(tableName);
        ArrayList<String> idsList = new ArrayList<>();

        Scan scan = new Scan();
        //创建列值过滤器
        ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("男")));

        scan.setFilter(valueFilter);

        ResultScanner scanner = students2.getScanner(scan);
        for (Result result : scanner) {
            String id = Bytes.toString(result.getRow());
            idsList.add(id);
        }
        return idsList;
    }


    public static List<String> getGirlIds(TableName tableName) throws Exception{
        Table students2 = conn.getTable(tableName);
        ArrayList<String> idsList = new ArrayList<>();

        Scan scan = new Scan();
        //创建列值过滤器
        ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("女")));

        scan.setFilter(valueFilter);

        ResultScanner scanner = students2.getScanner(scan);
        for (Result result : scanner) {
            String id = Bytes.toString(result.getRow());
            idsList.add(id);
        }
        return idsList;
    }

    public static void init() {
        //获取数据库的连接对象和数据库的操作对象
        ArrayList<Object> arrayList = MyHBaseTools.getConnection();
        conn = (Connection) arrayList.get(0);
        admin = (Admin) arrayList.get(1);

        //创建redis连接对象
        jedis = new Jedis("192.168.21.150",7000);
    }
}
